package org.example.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * @author Tang
 * @classname OrderProducer
 * @description [ 顺序消息发送 ]
 * @date 2022/11/27 15:11
 */
@Slf4j
public class OrderProducer {

    public static void main(String[] args) {
        Thread threadCosumer = new Thread(() -> {
            try {
                cosumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        threadCosumer.setDaemon(true);
        threadCosumer.start();
        producer();
    }
    public static void cosumer() throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("OrderProducer");
        consumer.setNamesrvAddr("localhost:9876");

        consumer.subscribe("TopicTest","*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.printf("\n %s Receive New Messages: %n", Thread.currentThread().getName());
                for (MessageExt msg : msgs) {
                    System.out.print("Topic: " + msg.getTopic() + ",QueueId: " + msg.getQueueId() + ",msg: " + new String(msg.getBody()) + " ");
                }
                System.out.println();
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        log.info("consumer.start...");
        System.in.read();
    }

    public static void producer(){
        try {
            DefaultMQProducer producer = new DefaultMQProducer("OrderProducer");
            producer.setNamesrvAddr("localhost:9876");
            producer.start();
            // 顺序消息发送时将ShardingKey相同（同一订单号）的消息序路由到一个逻辑队列中。
            String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
            for (int i = 0; i < 100; i++) {
                int orderId = i % 10;
                Message msg =
                        new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(msg, new MessageQueueSelector() {
                    // arg是上述send接口中传入的Object对象，返回的是该消息需要发送到的队列。
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        // orderId % mqs.size()
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId);

//                System.out.printf("%s%n", sendResult);
            }
            producer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
